package wsz.dlm;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;

/**
 * 消费失败-模拟死信
 * @author wsz
 * @date 2022/2/25 14:00
 **/
public class Consumer {

    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_grp_wsz");
        consumer.setNamesrvAddr("192.168.174.138:9876");
        // 主题订阅
        consumer.subscribe("tp_demo_06", "*");

        // 最多重试1次
        consumer.setMaxReconsumeTimes(1);
        /**
         * 无序
         * 消费失败进行重试，间隔1s
         */
        consumer.setMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg: msgs) {
                System.out.println(msg.getMsgId() + "\t" +
                        msg.getQueueId() + "\t" +
                        new String(msg.getBody()) + "  " + msg.getReconsumeTimes());
            }
            //返回null，死信
//            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            return null;
        });

        consumer.start();
    }
}
